package com.boarsoft.message.store;

import java.util.Date;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;

import com.boarsoft.common.util.DateUtil;
import com.boarsoft.message.bean.Message;
import com.boarsoft.message.bean.MessageEntity;
import com.boarsoft.message.bean.MessageQueue;
import com.boarsoft.message.bean.TargetMQ;
import com.boarsoft.message.core.MessageCore;

/**
 * 消息恢复器的默认实现
 * 
 * @author Mac_J
 *
 */
public class MessageRecoverImpl implements MessageRecover, Runnable {
	private static final Logger log = LoggerFactory.getLogger(MessageRecoverImpl.class);

	/** 自动注入 */
	@Autowired
	@Lazy
	protected MessageCore messageCore;

	/** 程序注入 */
	protected MessageRoller roller;
	/** 程序注入 */
	protected TargetMQ queue;

	/** Spring注入 */
	protected long interval = 3000L;
	/** Spring注入 */
	protected int limit = 1000;

	/** 供远程控制 */
	protected volatile boolean active = true;

	protected Thread thread = new Thread(this);

	@Override
	public void start() {
		thread.start();
	}

	@Override
	public void run() {
		// 不断检查是否有发送失败的消息
		int term = queue.getTerm();
		//
		Date since = new Date();
		while (true) {
			try {
				Thread.sleep(interval);
			} catch (InterruptedException e) {
				log.error("Be interrupted wait for next scan");
			}
			if (!active) {
				continue;
			}
			Date end = new Date();
			// 目前查询的是当前48小时以内的消息
			Date begin = DateUtil.addMinute(end, -term);
			int loaded = 0;
			// 本时段的所有消息都处理完成才继续
			do {
				// 如果begin早于服务器启动时间
				if (begin.before(since)) {
					// 依次装载begin和since之间，状态为NEW/QUEUED/SENDING的消息
					loaded = this.reload(begin, since, Message.STATUS_NEW);
					if (loaded > 0) {
						log.warn("Reloaded {} NEW messages between {} and {}", //
								loaded, begin.getTime(), end.getTime());
					}
					loaded = this.reload(begin, since, Message.STATUS_QUEUED);
					if (loaded > 0) {
						log.warn("Reloaded {} QUEUED messages between {} and {}", //
								loaded, begin.getTime(), end.getTime());
					}
					//
					loaded = this.reload(begin, since, Message.STATUS_SENDING);
					if (loaded > 0) {
						log.warn("Reloaded {} SENDING messages between {} and {}", //
								loaded, begin.getTime(), end.getTime());
					}
				}
				if (queue.getReliability() == TargetMQ.RA_TRAN) {
					// 装载服务器重启后，状态为失败的消息，此时begin>=since
					loaded = this.reload(begin, end, Message.STATUS_FAILED);
					if (loaded > 0) {
						log.warn("Reloaded {} FAILED messages between {} and {}", //
								loaded, begin.getTime(), end.getTime());
					}
				}
			} while (loaded > 0);
		}
	}

	protected int reload(Date begin, Date end, short status) {
		List<MessageEntity> meLt = null;
		try {
			meLt = roller.reload(begin, end, status, limit);
		} catch (Exception e) {
			log.error("Error on reload messages", e);
			return 0;
		}
		if (meLt == null || meLt.isEmpty()) {
			return 0;
		}
		// 将这些消息装载到队列中，全部添加完成才退出
		for (MessageEntity me : meLt) {
			while (true) {
				try {
					// 添加到队列，并signalAll通知await的线程
					if (messageCore.handle(queue, me)) {
						break; // 添加成功就退出当前while循环继续for循环
					}
				} catch (Exception e) {
					log.error("Error on reload message {} to queue {}", me, queue, e);
				}
				// 添加不成功表示 队列已满，稍后再试
				try {
					Thread.sleep(interval);
				} catch (InterruptedException e) {
					log.error("Be interrupted while wait for queue");
				}
			}
		}
		return meLt.size();
	}

	@Override
	public void setQueue(TargetMQ queue) {
		this.queue = queue;
	}

	@Override
	public void setRoller(MessageRoller messageRoller) {
		this.roller = messageRoller;
	}

	// -------------

	public MessageRoller getRoller() {
		return roller;
	}

	public MessageQueue getQueue() {
		return queue;
	}

	public long getInterval() {
		return interval;
	}

	public void setInterval(long interval) {
		this.interval = interval;
	}

	public int getLimit() {
		return limit;
	}

	public void setLimit(int limit) {
		this.limit = limit;
	}

	public boolean isActive() {
		return active;
	}

	public void setActive(boolean active) {
		this.active = active;
	}
}
